package com.burges.net.tableAPIAndSQL.sql.groupBy

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.TableEnvironment

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/25 
  * 描述      窗口操作之HOPWindows使用代码示例
  */
object HOPWindowsDEMO {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tableEnvironment = TableEnvironment.getTableEnvironment(environment)
		//创建数据集
		val ds: DataStream[(Long, String, Int)] = environment.fromElements((1l, "flink", 1))
		val ds: DataStream[(Long, String, Int)] = environment.fromElements((1l, "flink", 1))
		//注册表名信息并定义字段proctime为ProcessTime，定义字段 rowtime为rowtime
		tableEnvironment.registerDataStream("Sensors", ds, 'id, 'type, 'var1, 'proctime.proctime, 'rowtime.rowtime)
		//基于proctime创建HOP窗口，并制定敞口长度为10min，每隔1min滑动一次窗口
		//根据id进行聚合求取var1的和
		tableEnvironment.sqlQuery("SELECT id, sum(var1) from Sensors group by HOP(proctime, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE), id")
		//基于rowtime创建HOP窗口，并制定窗口长度为10min，每隔1min滑动一次窗口
		//根据id进行聚合求取var1的和
		tableEnvironment.sqlQuery("SELECT id, sum(var1) from Sensors group by HOP(rowtime, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE), id")
		tableEnvironment.sqlQuery("SELECT id, " +
				//获取窗口起始时间并标记为wStart字段
				"HOP_START(rowtime, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE) as wStart " +
				//获取窗口结束时间并标记为wEnd字段
				"HOP_END(rowtime, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE) as wEnd " +
				"from Sensors group by HOP(rowtime, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE), id")
	}

}
